package edu.kit.aifb.ldstream.learner.tasks.mod;

import java.util.Set;

import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.repository.RepositoryException;
import org.openrdf.repository.RepositoryResult;
import org.semanticweb.yars.nx.namespace.RDF;


import edu.kit.aifb.ldstream.env.Environment;
import edu.kit.aifb.ldstream.stat.util.IDistributionChecker;

public class LearnClasses extends LearnerTask{

	private Set<Type> typeSet;
	private ModStatistics stats;
	private IDistributionChecker dc;

	public LearnClasses(Set<Type> typeSet, ModStatistics stats, IDistributionChecker dc) {
		this.typeSet = typeSet;
		this.dc = dc;
		this.stats = stats;
	}

	@Override
	public void run() {

		startup();

		if(Environment.LOGGING) Environment.log.info("Started to learn the Classes.");
		learnTypes();
		if(Environment.LOGGING) Environment.log.info("Finished learning the Classes.");

		if(Environment.LOGGING) Environment.log.info("Started to learn the Instance distribution.");
		sequentialInstanceDistribution();
		//instanceDistribution();
		if(Environment.LOGGING) Environment.log.info("Finished learning the instance distribution.");

		if(Environment.LOGGING)Environment.log.info("Started to learn the bandwidth distribution.");
		leanBandwidthDistribSequential();
		//	learnBandwidthDistrib();
		if(Environment.LOGGING) Environment.log.info("Finished learning the bandwidth distribution.");

		shutdown();
	}


	private void leanBandwidthDistribSequential(){

		int[] tripleCounts = new int[this.stats.getNumEvents()];
		RepositoryResult<Statement> result = null;
		try {
			result = this.con.getStatements(null, null, null, false);
			Statement s;
			int id;
			try{
				while(result.hasNext()){
					s = result.next();
					//retrieving the context of that triple. The local name is the EventID
					if(s.getContext() != null){
						id = Integer.parseInt(((URI) s.getContext()).getLocalName());
						tripleCounts[id]++;
					}
				}
			} finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		}
		Integer[] values = new Integer[this.stats.getNumEvents()];
		for(int i = 0; i < values.length; i++){
			values[i] = Integer.valueOf(tripleCounts[i]);
		}
		this.stats.setBandwidthDistribution(this.dc.discrete(values, false));
		System.gc();
	}

	/*
	 * Use leanBandwidthDistribSequential() instead!
	 */
	@Deprecated
	private void learnBandwidthDistrib(){
		String select = "SELECT (COUNT(?pred) AS ?numTriples) ";
		String where = " WHERE {?subj ?pred ?obj}";
		StringBuilder query;
		StringBuilder from; 
		Integer[] bandwidths = new Integer[this.stats.getNumEvents()];

		int tenth = Math.max(this.stats.getNumEvents() / 10, 1);
		for(int i = 0; i < this.stats.getNumEvents(); i++){
			if((i % tenth) == 0){
				System.gc();
				if(Environment.LOGGING) Environment.log.info("Bandwith distribution finished to " + (double) i / ((double) this.stats.getNumEvents())) ; 
			}
			from = new StringBuilder(" FROM <").append(Environment.EVENT_CONTEXT).append(i).append("> ");
			query = new StringBuilder(select).append(from).append(where);
			bandwidths[i] = Integer.valueOf(getCount(query.toString(), "numTriples"));
		}
		this.stats.setBandwidthDistribution(this.dc.discrete(bandwidths, false));
	}

	private void learnTypes(){
		/*
		String query = "SELECT DISTINCT ?type FROM <" + Environment.STREAM_CONTEXT_URI.stringValue() + "> "
						+ "WHERE { ?instance " + RDF.TYPE.toN3() + " ?type FILTER( !isBLANK(?instance)) }";
		 */
		StringBuilder query = new StringBuilder("SELECT DISTINCT ?type WHERE { ?instance ")
		.append(RDF.TYPE.toN3())
		.append(" ?type FILTER( !isBLANK(?instance)) }");
		try {
			TupleQuery tupleQuery = con.prepareTupleQuery(QueryLanguage.SPARQL, query.toString());
			TupleQueryResult result = tupleQuery.evaluate();
			try{
				BindingSet set;
				while(result.hasNext()){
					set = result.next();
					this.typeSet.add(new Type(set.getValue("type").stringValue()));
				}
			}finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
		System.gc();
	}

	private void sequentialInstanceDistribution(){
		RepositoryResult<Statement> result;
		for(Type type : this.typeSet){
			int[] instanceOcc = new int[this.stats.getNumEvents()];
			try {
				result = this.con.getStatements(null, org.openrdf.model.vocabulary.RDF.TYPE, type, false);
				Statement s;
				int eventID;
				try{
					while(result.hasNext()){
						s = result.next();
						//should not be a bnode
						if(s.getSubject() instanceof URI){
							//local name is the event id 
							eventID = Integer.parseInt(((URI) s.getContext()).getLocalName());
							instanceOcc[eventID]++;
						}
					}
				} finally{
					result.close();
				}
			} catch (RepositoryException e) {
				e.printStackTrace();
			}
			Integer[] temp = new Integer[this.stats.getNumEvents()];
			for(int i = 0; i < instanceOcc.length; i++){
				temp[i] = Integer.valueOf(instanceOcc[i]);
			}
			type.setInstanceOcc(this.dc.discrete(temp, false));
			// free memory
			temp = null;
			instanceOcc = null;
			System.gc();
		}
	}

	@Deprecated
	private void instanceDistribution(){

		Integer[] occurrences;
		StringBuilder from;
		StringBuilder where;
		StringBuilder query;
		// Select clause (type and event independent)
		String select = "SELECT (COUNT(?x) AS ?numInstances) ";
		for(Type type : this.typeSet){
			occurrences = new Integer[this.stats.getNumEvents()];
			//Where clause (type dependent)
			where = new StringBuilder(" WHERE {?x ").append(RDF.TYPE.toN3()).append(" <").append(type.stringValue()).append("> FILTER(!isBLANK(?x))}");
			for(int i = 0; i < this.stats.getNumEvents(); i++){
				//build from clause
				from = new StringBuilder(" FROM <").append(Environment.EVENT_CONTEXT).append(i).append("> ");
				// build query
				query = new StringBuilder(select).append(from).append(where);
				occurrences[i] = Integer.valueOf(getCount(query.toString(), "numInstances"));
			}
			type.setInstanceOcc(this.dc.discrete(occurrences, false));
			occurrences = null;
		}

		System.gc();
	}

	private int getCount(String query, String variableName){
		int count = 0;
		try {
			TupleQuery tq= this.con.prepareTupleQuery(QueryLanguage.SPARQL, query);
			TupleQueryResult result = tq.evaluate();
			try{
				if(result.hasNext()){
					count = Integer.parseInt(result.next().getValue(variableName).stringValue());
				}
			}finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
		return count;
	}


}
